package com.niit.dao;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

/**
 * @author 杨铭
 * 2022/2/10,22:23
 */
//实现使用txt文件批量向HBase导入数据
public class PutAll {
    //读取学生的全部信息并写入之前创建的student表中



    public void putAll() throws IOException {
        HbaseConnection hbaseConnection =new HbaseConnection();
        Configuration conf = hbaseConnection.connect();
        //读取学生信息
        //改成自己的文件路径
        BufferedReader br = new BufferedReader(new FileReader("C:\\Users\\26474\\Desktop\\其他\\大二项目\\pro2.txt"));
        //与hbase里面的student表建立连接
//        Table student = conn.getTable(TableName.valueOf("student"));
        HTable student=new HTable(conf,"Student_Table");
        String line = null;
        //创建put的集合，用于批量插入数据，避免每次循环都执行一次put语句
        ArrayList<Put> puts = new ArrayList<>();
        int batchSize = 10;//以十条数据为一组批量导入
        while((line=br.readLine())!=null){
            //写入hbase里面的student表
            String[] split = line.split("\t");
            String studentID = split[0];
            String name = split[1];
           // String  dateOfEnrollment= split[2];
            String contactNumber = split[2];
            String semester = split[3];
            String studentStatus=split[4];

            //往put对象里面放入rowkey
            Put put = new Put(studentID.getBytes());
            //往put对象里面放入value
            //addColumn()括号里面的参数依次是列簇，列名，值
            byte[] info = "info".getBytes();
            byte [] status="status".getBytes();
            put.addColumn(info,"Name".getBytes(),name.getBytes());
            put.addColumn(info,"ContactNumber".getBytes(),contactNumber.getBytes());
            put.addColumn(status,"Semester".getBytes(),semester.getBytes());
            put.addColumn(status,"StudentStatus".getBytes(),studentStatus.getBytes());

            //批量执行,将当前循环的新的put对象放入puts集合中
            puts.add(put);
            //当puts集合的大小和之前设定的batchSize大小一致时
            //利用getTable获取的student对象来调用put方法，将puts里面的数据写入student表
            //注意不要混淆，利用put方法，将puts集合中的多个put对象的数据写入student表
            if(puts.size() == batchSize){
                student.put(puts);
                //集合里面的数据都写入完后，清空集合，准备迎接下一批次的数据
                puts.clear();
            }
        }
        System.out.println(puts.isEmpty());
        System.out.println(puts.size());
        //这里还要注意一件事情，那就是students文件的数据总数是batchSize的整数倍时才能全部写入
        //否则会有几条语句因为不满batchSize的大小，没有触发student的put方法
        //因此在这里再加上一个判断，如果还有几条数据没有写入，那么写入
        if(!puts.isEmpty()){
            student.put(puts);
        }
        //关闭br
        br.close();
    }

    public static void main(String[] args) throws IOException{
        PutAll putAll=new PutAll();
        putAll.putAll();
    }

}
